package com.joker.aigc.chat.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.joker.aigc.chat.domain.dto.WebSocketChannelExtraDTO;
import com.joker.aigc.config.ThreadPoolConfig;
import com.joker.aigc.domain.entity.User;
import com.joker.aigc.chat.domain.entity.websocket.WebSocketAuthorize;
import com.joker.aigc.chat.domain.entity.websocket.WebSocketBaseVo;

import com.joker.aigc.domain.enums.RoleEnum;
import com.joker.aigc.service.LoginService;
import com.joker.aigc.chat.service.SocketService;
import com.joker.aigc.service.UserRoleService;
import com.joker.aigc.service.UserService;
import com.joker.aigc.chat.service.adapter.WebSocketAdapter;
import com.joker.aigc.chat.service.event.UserOfflineEvent;
import com.joker.aigc.chat.service.event.UserOnlineEvent;
import com.joker.aigc.utils.NettyUtil;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Date;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;


@Service
@Slf4j
public class SocketServiceImpl implements SocketService {

    // 这里用来处理所有 关于socket的任务。比如 用户的上线，下线。还有发送

    private static final Duration EXPIRE_TIME = Duration.ofHours(1);
    private static final Long MAX_MUM_SIZE = 10000L;
    /**
     * 所有请求登录的code与channel关系
     */
    public static final Cache<Integer, Channel> WAIT_LOGIN_MAP = Caffeine.newBuilder()
            .expireAfterWrite(EXPIRE_TIME)
            .maximumSize(MAX_MUM_SIZE)
            .build();
    /**
     * 所有已连接的websocket连接列表和一些额外参数
     */
    private static final ConcurrentHashMap<Channel, WebSocketChannelExtraDTO> ONLINE_CHANNEL_MAP = new ConcurrentHashMap<>();
    /**
     * 所有在线的用户和对应的socket
     */
    public static final ConcurrentHashMap<Long, CopyOnWriteArrayList<Channel>> ONLINE_UID_MAP = new ConcurrentHashMap<>();

    public static ConcurrentHashMap<Channel, WebSocketChannelExtraDTO> getOnlineMap() {
        return ONLINE_CHANNEL_MAP;
    }


    @Autowired
    @Qualifier(ThreadPoolConfig.WS_EXECUTOR)
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    @Autowired
    private LoginService loginService;


    @Autowired
    private UserService userService;

    @Autowired
    private UserRoleService userRoleService;

    /**
     * 如果有token信息的话，则进行验证
     *
     * @param channel     websocket 的通道信息
     * @param wsAuthorize 校验信息
     */
    @Override
    public void authorize(Channel channel, WebSocketAuthorize wsAuthorize) {
        log.info("login step 1: 开始认证 token信息 authorize-->{}", wsAuthorize.getToken());
        //校验token
        boolean verifySuccess = loginService.verify(wsAuthorize.getToken());
        if (verifySuccess) {//用户校验成功给用户登录
            // 获取用户。校验是否正确
            User user = userService.getById(loginService.getValidUid(wsAuthorize.getToken()));
            log.info("login step 2: 认证成功 查询到的user的信息-->{}", user);
            // 用户登录成功够，就会进行各种操作
            loginSuccess(channel, user, wsAuthorize.getToken());
        } else {
            log.info("login step 3: 认证失败 token信息失效，给前端发送重新获取 token的消息。type=6");
            //让前端的token失效
            sendMsg(channel, WebSocketAdapter.buildInvalidateTokenResp());
        }
    }


    private void loginSuccess(Channel channel, User user, String token) {
        //更新上线列表
        online(channel, user.getId());
        log.info("login step 2-1 成功登录--更新channel通道信息");
        //返回给用户登录成功
        boolean hasPower = userRoleService.hasPower(user.getId(), RoleEnum.ADMIN);
        //发送给对应的用户
        sendMsg(channel, WebSocketAdapter.buildLoginSuccessResp(user, token, hasPower));
        //发送用户上线事件
        Boolean online = userService.isOnline(user.getId());
        // 不需要判断 是否已经在线了
        log.info("login step 2-2 成功登录--通知其他在线用户，该用户上线");
        user.setLastOptTime(new Date());
        user.refreshIp(NettyUtil.getAttr(channel, NettyUtil.IP));
        applicationEventPublisher.publishEvent(new UserOnlineEvent(this, user));

    }

    @Override
    public void sendToAllOnline(WebSocketBaseVo<?> wsBaseResp, Long skipUid) {
        // 发送给所有人。然后跳过那些人
        log.info("全局消息，忽略的人:{} 发送给所有人:{},", skipUid, wsBaseResp);
        ONLINE_CHANNEL_MAP.forEach((channel, ext) -> {
            if (Objects.nonNull(skipUid) && Objects.equals(ext.getUid(), skipUid)) {
                return;
            }
            threadPoolTaskExecutor.execute(() -> sendMsg(channel, wsBaseResp));
        });
    }


    @Override
    public void removed(Channel channel) {
        // 判断 用户的uid是否存在
        WebSocketChannelExtraDTO wsChannelExtraDTO = ONLINE_CHANNEL_MAP.get(channel);
        Optional<Long> uidOptional = Optional.ofNullable(wsChannelExtraDTO).map(WebSocketChannelExtraDTO::getUid);
        boolean offlineAll = offline(channel, uidOptional);
        if (uidOptional.isPresent() && offlineAll) {//已登录用户断连,并且全下线成功
            User user = new User();
            user.setId(uidOptional.get());
            user.setLastOptTime(new Date());
            // 这里是发布事件。让其他线程去处理 下线任务时间
            applicationEventPublisher.publishEvent(new UserOfflineEvent(this, user));
        }
    }

    // 上线
    private void online(Channel channel, Long uid) {
        // step 1:更新用户通道
        getOrInitChannelExt(channel).setUid(uid);
        ONLINE_UID_MAP.putIfAbsent(uid, new CopyOnWriteArrayList<>());
        ONLINE_UID_MAP.get(uid).add(channel);
        NettyUtil.setAttr(channel, NettyUtil.UID, uid);
    }


    private WebSocketChannelExtraDTO getOrInitChannelExt(Channel channel) {
        //
        WebSocketChannelExtraDTO wsChannelExtraDTO = ONLINE_CHANNEL_MAP.getOrDefault(channel, new WebSocketChannelExtraDTO());
        WebSocketChannelExtraDTO old = ONLINE_CHANNEL_MAP.putIfAbsent(channel, wsChannelExtraDTO);
        return ObjectUtil.isNull(old) ? wsChannelExtraDTO : old;
    }


    /**
     * 用户下线
     * return 是否全下线成功
     */
    private boolean offline(Channel channel, Optional<Long> uidOptional) {
        // step 1: 下线首先就是移除 channel
        ONLINE_CHANNEL_MAP.remove(channel);
        if (uidOptional.isPresent()) {
            // 如果
            CopyOnWriteArrayList<Channel> channels = ONLINE_UID_MAP.get(uidOptional.get());
            if (CollUtil.isNotEmpty(channels)) {
                channels.removeIf(ch -> Objects.equals(ch, channel));
            }
            return CollUtil.isEmpty(ONLINE_UID_MAP.get(uidOptional.get()));
        }
        return true;
    }


    /**
     * 处理所有ws连接的事件
     *
     * @param channel
     */
    @Override
    public void connect(Channel channel) {
        ONLINE_CHANNEL_MAP.put(channel, new WebSocketChannelExtraDTO());
    }


    @Override
    public void sendToUid(WebSocketBaseVo<?> wsBaseResp, Long uid) {
        log.info("sendToUid->uid:{},wsBase:{}", uid, wsBaseResp);
        if (uid == null) {
            return;
        }
        CopyOnWriteArrayList<Channel> channels = ONLINE_UID_MAP.get(uid);
        if (CollUtil.isEmpty(channels)) {
            log.info("用户：{}不在线", uid);
            return;
        }
        channels.forEach(channel -> threadPoolTaskExecutor.execute(() -> sendMsg(channel, wsBaseResp)));
    }

    @Override
    public void sendMsg(Channel channel, WebSocketBaseVo<?> wsBaseResp) {
        /**
         * 发送各种消息
         */
        channel.writeAndFlush(new TextWebSocketFrame(JSONUtil.toJsonStr(wsBaseResp)));
    }


}
